package com.hup.utils.io.socket.udp;

import com.hup.utils.commons.exception.AppCommonException;
import com.hup.utils.commons.exception.AppExceptionHandler;
import com.hup.utils.commons.exception.ErrInfo;
import com.hup.utils.commons.thread.ThreadUtil;
import lombok.Getter;
import lombok.extern.log4j.Log4j2;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.util.concurrent.ThreadPoolExecutor.AbortPolicy;

/**
 * 关于UDP
 * <pre>
 *     .UDP本身无server/client之分,
 *      都是DatagramSocket,启动时都占用本机的一个端口号,发送消息时都要指定目标的address和port
 *     .发送消息时,如果目标没启动,则不会有异常
 *     .初始化时,如果传入的port为null,则随机占用端口号(程序关闭时会自动释放)
 * </pre>
 * 查询占用端口号: netstat -nao | findstr ":3306"
 * @author hugan
 * @date 2020/3/29
 */
@Log4j2
public class UdpCore {

    /**
     * 构造时的端口号
     */
    private final Integer defaultPort;
    /**
     * 本机占用的端口号;
     * 如果构造时为null,则在首次触发[读取线程]或[发送数据]时被刷新
     */
    @Getter
    private Integer port;
    @Getter
    private final UdpHandler handler;
    private final ThreadPoolTaskExecutor executor;

    private boolean hadStarted;
    private boolean canRun;

    private DatagramSocket socket;
    private DupDataReceiver receiver;

    /**
     * 只负责发送消息;true-不启动接受数据的线程
     */
    @Getter
    private boolean onlySend;


    protected UdpCore(UdpHandler handler) {
        this(null, handler);
    }

    protected UdpCore(Integer port, UdpHandler handler) {
        defaultPort = this.port = port;
        this.handler = handler;
        executor = initExecutor();
    }

    protected ThreadPoolTaskExecutor initExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(0);//核心线程池数量
        executor.setMaxPoolSize(2);
        executor.setQueueCapacity(0);
        executor.setThreadNamePrefix("udp-");
        executor.setRejectedExecutionHandler(new AbortPolicy());//超出任务数时抛异常,不会发生的
        executor.setKeepAliveSeconds(0);
        executor.initialize();
        return executor;
    }

    /**
     * 设置下次建立连接时,是否不启动[接受消息线程]
     */
    public void setOnlySend(boolean onlySend) {
        if (hadStarted) {
            throw new IllegalArgumentException("请先断开连接");
        }
        this.onlySend = onlySend;
    }

    public void start() {
        if(hadStarted) {
            throw new AppCommonException("服务器已启动");
        }

        try {
            if (port == null) {
                socket = new DatagramSocket(null);
            } else {
                socket = new DatagramSocket(port);
                onGetLocalPort();
            }
            receiver = new DupDataReceiver();
            handler.onStarted(true, null);
            hadStarted = true;
            canRun = true;

            if (!onlySend) startReceiveThread();
        } catch (Exception e) {
            ErrInfo errInfo = AppExceptionHandler.handle("UDP启动失败:", e);
            handler.onStarted(false, errInfo);
            hadStarted = false;
            canRun = false;
        }
    }

    private void startReceiveThread() {
        if (defaultPort == null) {
            executor.execute(() -> {
                //要在socket.receive(inputPacket);调用后才回获取到端口号
                ThreadUtil.sleep(500);
                onGetLocalPort();
            });
        }
        executor.execute(() -> {
            try {
                while (canRun) {
                    //等待读取数据
                    UdpData data = receiver.receive(socket);
                    handler.onReceive(data);
                }
            } catch (IOException e) {
                hadStarted = false;
                ErrInfo errInfo = AppExceptionHandler.handleNoLog("", e);
                if (!canRun) {
                    log.debug("udp主动关闭:{}", errInfo.getErrMsg());
                    handler.onStopped(true, null);
                } else {
                    log.info("udp异常关闭:{}", errInfo.getErrMsg());
                    handler.onStopped(false, errInfo);
                }
            }
        });
    }

    private void onGetLocalPort() {
        port = socket.getLocalPort();
        log.info("port={}", port);
    }

    public void stop() {
        try {
            canRun = false;
            if (onlySend) {
                hadStarted = false;
                //只发时,没有read线程,不会在断开后更新[hadStarted]
            }
            if (socket != null) {
                //[hadStarted]在断开后会被更新
                socket.close();
                socket = null;
            }
            port = defaultPort;
        } catch (Exception e) {
            //没遇到过,如果server已关闭,重复调用没异常
            AppExceptionHandler.handle("udp关闭异常:", e);
        }
    }

    /**
     * 发送消息,异步的
     */
    public void send(byte[] data, InetAddress address, int port, SendDataHandler handler) {
        executor.execute(() -> {
            try {
                DatagramPacket packet = new DatagramPacket(data, data.length, address, port);
                UdpCore.this.handler.beforeSend(packet);
                socket.send(packet);
                if (defaultPort == null && onlySend) {
                    onGetLocalPort();
                }
            } catch (Exception e) {
                handler.onException(e);
            }
        });
    }

    public interface UdpHandler {

        /**
         * 通知启动结果
         */
        void onStarted(boolean success, ErrInfo errInfo);

        /**
         * 关闭结果
         * @param byHand true-正常人工关闭;false-由于异常被动关闭
         * @param errInfo notNull:byHand==false
         */
        void onStopped(boolean byHand, ErrInfo errInfo);

        default void beforeSend(DatagramPacket packet) {}

        default void onReceive(UdpData data) {}

    }

    public interface SendDataHandler {
        void onException(Exception e);
    }

}
